package com.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSONObject;
import com.gmall.realtime.common.base.BaseApp;
import com.gmall.realtime.common.bean.TrafficHomeDetailPageViewBean;
import com.gmall.realtime.common.constant.Constant;
import com.gmall.realtime.common.function.DorisMapFunction;
import com.gmall.realtime.common.util.DateFormatUtil;
import com.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficHomeDetailPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficHomeDetailPageViewWindow().start(
                10023,
                1,
                "dws_traffic_home_detail_page_view_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
/**
 * 处理数据流的主方法
 * @param env Flink流执行环境
 * @param dataStreamSource 输入数据流，包含JSON格式的页面访问数据
 */
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 1. 数据清洗：过滤出首页(home)和商品详情页(good_detail)的访问记录
        SingleOutputStreamOperator<JSONObject> etlStream = dataStreamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String mid = jsonObject.getJSONObject("common").getString("mid"); // 设备ID
                String ts = jsonObject.getString("ts"); // 时间戳
                String pageId = jsonObject.getJSONObject("page").getString("page_id"); // 页面ID

                // 只保留首页和商品详情页的记录，且mid和ts不为空
                if ("home".equals(pageId) || "good_detail".equals(pageId)) {
                    if (mid != null && ts != null) {
                        collector.collect(jsonObject);
                    }
                }
            }
        });

        // 2. 按设备ID(mid)分组，统计每个设备的首页和详情页UV
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> keyStream = etlStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        }).process(new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {
            private ValueState<String> homeState;   // 记录设备访问首页的最后日期
            private ValueState<String> detailState; // 记录设备访问详情页的最后日期

            @Override
            public void open(Configuration parameters) throws Exception {
                // 初始化状态，设置TTL为1天
                ValueStateDescriptor<String> homeStateDesc = new ValueStateDescriptor<>("home_state", String.class);
                ValueStateDescriptor<String> detailStateDesc = new ValueStateDescriptor<>("detail_state", String.class);
                homeStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                detailStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                homeState = getRuntimeContext().getState(homeStateDesc);
                detailState = getRuntimeContext().getState(detailStateDesc);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>.Context context,
                                       Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {
                // 获取当前记录的日期和页面类型
                String homeValue = homeState.value();
                String detailValue = detailState.value();
                String curDt = DateFormatUtil.tsToDate(jsonObject.getLong("ts"));
                String pageId = jsonObject.getJSONObject("page").getString("page_id");

                long homeCt = 0L;   // 首页UV计数
                long detailCt = 0L; // 详情页UV计数

                // 判断是否是当天首次访问首页
                if ("home".equals(pageId)) {
                    if (!curDt.equals(homeValue)) {
                        homeCt = 1L;
                        homeState.update(curDt);
                    }
                }
                // 判断是否是当天首次访问详情页
                else {
                    if (!curDt.equals(detailValue)) {
                        detailCt = 1L;
                        detailState.update(curDt);
                    }
                }

                // 如果当天有首次访问，则输出统计结果
                if (homeCt !=0 || detailCt != 0){
                    collector.collect(TrafficHomeDetailPageViewBean.builder()
                            .homeUvCt(homeCt)
                            .goodDetailUvCt(detailCt)
                            .ts(jsonObject.getLong("ts"))
                            .build());
                }
            }
        });

        // 3. 添加水位线，开10秒的滚动窗口进行聚合
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> reduceStream = keyStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<TrafficHomeDetailPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficHomeDetailPageViewBean>() {
                                    @Override
                                    public long extractTimestamp(TrafficHomeDetailPageViewBean trafficHomeDetailPageViewBean, long l) {
                                        return trafficHomeDetailPageViewBean.getTs();
                                    }
                                }))
                .windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(
                        // 窗口内数据聚合：累加首页和详情页UV
                        new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                            @Override
                            public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1, TrafficHomeDetailPageViewBean value2) throws Exception {
                                value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                                value1.setGoodDetailUvCt(value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt());
                                return value1;
                            }
                        },
                        // 窗口处理：添加窗口时间信息
                        new ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
                            @Override
                            public void process(Context context, Iterable<TrafficHomeDetailPageViewBean> iterable,
                                                Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {
                                TimeWindow window = context.window();
                                String start = DateFormatUtil.tsToDateTime(window.getStart()); // 窗口开始时间
                                String end = DateFormatUtil.tsToDateTime(window.getEnd());     // 窗口结束时间

                                TrafficHomeDetailPageViewBean bean = iterable.iterator().next();
                                bean.setStt(start);
                                bean.setEdt(end);
                                bean.setCurDate(DateFormatUtil.tsToDate(System.currentTimeMillis()));
                                collector.collect(bean);
                            }
                        });

        // 4. 将结果写入Doris数据库
        reduceStream
                .map(new DorisMapFunction<>())
                .sinkTo(
                        FlinkSinkUtil.getDorisSink(
                                Constant.DORIS_DATABASE + ".dws_traffic_home_detail_page_view_window",
                                "dws_traffic_home_detail_page_view_window"
                        )
                );
    }


}